/*-------------------------------------------------------------------------
 *
 * blvalidate.c
 *	  Opclass validator for bloom.
 *
 * Copyright (c) 2016-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  contrib/bloom/blvalidate.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/amvalidate.h"
#include "access/htup_details.h"
#include "bloom.h"
#include "catalog/pg_amop.h"
#include "catalog/pg_amproc.h"
#include "catalog/pg_opclass.h"
#include "catalog/pg_opfamily.h"
#include "catalog/pg_type.h"
#include "utils/builtins.h"
#include "utils/lsyscache.h"
#include "utils/regproc.h"
#include "utils/syscache.h"

/*
 * Bloom操作类的验证器。
 */
bool blvalidate(Oid fc_opclassoid)
{
	bool		fc_result = true;
	HeapTuple	fc_classtup;
	Form_pg_opclass fc_classform;
	Oid			fc_opfamilyoid;
	Oid			fc_opcintype;
	Oid			fc_opckeytype;
	char	   *fc_opclassname;
	HeapTuple	fc_familytup;
	Form_pg_opfamily fc_familyform;
	char	   *fc_opfamilyname;
	CatCList   *fc_proclist,
			   *fc_oprlist;
	List	   *fc_grouplist;
	OpFamilyOpFuncGroup *fc_opclassgroup;
	int			fc_i;
	ListCell   *fc_lc;

	/* 获取操作类信息 */
	fc_classtup = SearchSysCache1(CLAOID, ObjectIdGetDatum(fc_opclassoid));
	if (!HeapTupleIsValid(fc_classtup))
		elog(ERROR, "cache lookup failed for operator class %u", fc_opclassoid);
	fc_classform = (Form_pg_opclass) GETSTRUCT(fc_classtup);

	fc_opfamilyoid = fc_classform->opcfamily;
	fc_opcintype = fc_classform->opcintype;
	fc_opckeytype = fc_classform->opckeytype;
	if (!OidIsValid(fc_opckeytype))
		fc_opckeytype = fc_opcintype;
	fc_opclassname = NameStr(fc_classform->opcname);

	/* 获取操作族信息 */
	fc_familytup = SearchSysCache1(OPFAMILYOID, ObjectIdGetDatum(fc_opfamilyoid));
	if (!HeapTupleIsValid(fc_familytup))
		elog(ERROR, "cache lookup failed for operator family %u", fc_opfamilyoid);
	fc_familyform = (Form_pg_opfamily) GETSTRUCT(fc_familytup);

	fc_opfamilyname = NameStr(fc_familyform->opfname);

	/* 获取操作族的所有操作符和支持函数 */
	fc_oprlist = SearchSysCacheList1(AMOPSTRATEGY, ObjectIdGetDatum(fc_opfamilyoid));
	fc_proclist = SearchSysCacheList1(AMPROCNUM, ObjectIdGetDatum(fc_opfamilyoid));

	/* 检查单个支持函数 */
	for (fc_i = 0; fc_i < fc_proclist->n_members; fc_i++)
	{
		HeapTuple	fc_proctup = &fc_proclist->members[fc_i]->tuple;
		Form_pg_amproc fc_procform = (Form_pg_amproc) GETSTRUCT(fc_proctup);
		bool		fc_ok;

		/*
		 * 所有的bloom支持函数应该与匹配的
		 * 左/右类型注册
		 */
		if (fc_procform->amproclefttype != fc_procform->amprocrighttype)
		{
			ereport(INFO,
					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
					 errmsg("bloom opfamily %s contains support procedure %s with cross-type registration",
							fc_opfamilyname,
							format_procedure(fc_procform->amproc))));
			fc_result = false;
		}

		/*
		 * 我们不能检查签名，除非在特定的操作类内，因为
		 * 在很多情况下我们需要知道相关的 opckeytype。
		 */
		if (fc_procform->amproclefttype != fc_opcintype)
			continue;

		/* 检查过程编号和函数签名 */
		switch (fc_procform->amprocnum)
		{
			case BLOOM_HASH_PROC:
				fc_ok = check_amproc_signature(fc_procform->amproc, INT4OID, false,
											1, 1, fc_opckeytype);
				break;
			case BLOOM_OPTIONS_PROC:
				fc_ok = check_amoptsproc_signature(fc_procform->amproc);
				break;
			default:
				ereport(INFO,
						(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
						 errmsg("bloom opfamily %s contains function %s with invalid support number %d",
								fc_opfamilyname,
								format_procedure(fc_procform->amproc),
								fc_procform->amprocnum)));
				fc_result = false;
				continue;		/* 不想要额外的消息 */
		}

		if (!fc_ok)
		{
			ereport(INFO,
					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
					 errmsg("gist opfamily %s contains function %s with wrong signature for support number %d",
							fc_opfamilyname,
							format_procedure(fc_procform->amproc),
							fc_procform->amprocnum)));
			fc_result = false;
		}
	}

	/* 检查单个操作符 */
	for (fc_i = 0; fc_i < fc_oprlist->n_members; fc_i++)
	{
		HeapTuple	fc_oprtup = &fc_oprlist->members[fc_i]->tuple;
		Form_pg_amop fc_oprform = (Form_pg_amop) GETSTRUCT(fc_oprtup);

		/* 检查这是bloom允许的策略 */
		if (fc_oprform->amopstrategy < 1 ||
			fc_oprform->amopstrategy > BLOOM_NSTRATEGIES)
		{
			ereport(INFO,
					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
					 errmsg("bloom opfamily %s contains operator %s with invalid strategy number %d",
							fc_opfamilyname,
							format_operator(fc_oprform->amopopr),
							fc_oprform->amopstrategy)));
			fc_result = false;
		}

		/* bloom不支持ORDER BY操作符 */
		if (fc_oprform->amoppurpose != AMOP_SEARCH ||
			OidIsValid(fc_oprform->amopsortfamily))
		{
			ereport(INFO,
					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
					 errmsg("bloom opfamily %s contains invalid ORDER BY specification for operator %s",
							fc_opfamilyname,
							format_operator(fc_oprform->amopopr))));
			fc_result = false;
		}

		/* 检查操作符签名 --- 所有bloom策略相同 */
		if (!check_amop_signature(fc_oprform->amopopr, BOOLOID,
								  fc_oprform->amoplefttype,
								  fc_oprform->amoprighttype))
		{
			ereport(INFO,
					(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
					 errmsg("bloom opfamily %s contains operator %s with wrong signature",
							fc_opfamilyname,
							format_operator(fc_oprform->amopopr))));
			fc_result = false;
		}
	}

	/* 现在检查操作符/函数组的不一致性 */
	fc_grouplist = identify_opfamily_groups(fc_oprlist, fc_proclist);
	fc_opclassgroup = NULL;
	foreach(fc_lc, fc_grouplist)
	{
		OpFamilyOpFuncGroup *fc_thisgroup = (OpFamilyOpFuncGroup *) lfirst(fc_lc);

		/* 记住与测试操作类完全匹配的组 */
		if (fc_thisgroup->lefttype == fc_opcintype &&
			fc_thisgroup->righttype == fc_opcintype)
			fc_opclassgroup = fc_thisgroup;

		/*
		 * 我们能做的检查操作符集合的事情不多，因为每个
		 * bloom操作类或多或少是独立的，并且有些只包含与操作类数据类型
		 * 兼容的操作符（这意味着空的操作符集合是可以的）。这种情况也意味着
		 * 除了操作类自己的组以外，我们不应该坚持非空函数集合。
		 */
	}

	/* 检查原名操作类是否完整 */
	for (fc_i = 1; fc_i <= BLOOM_NPROC; fc_i++)
	{
		if (fc_opclassgroup &&
			(fc_opclassgroup->functionset & (((uint64) 1) << fc_i)) != 0)
			continue;			/* 明白了 */
		if (fc_i == BLOOM_OPTIONS_PROC)
			continue;			/* 可选方法 */
		ereport(INFO,
				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
				 errmsg("bloom opclass %s is missing support function %d",
						fc_opclassname, fc_i)));
		fc_result = false;
	}

	ReleaseCatCacheList(fc_proclist);
	ReleaseCatCacheList(fc_oprlist);
	ReleaseSysCache(fc_familytup);
	ReleaseSysCache(fc_classtup);

	return fc_result;
}
